package com.bw.gmall.realtime.app.dws.xz;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficItemPvUvBean;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 计算商品PV/UV
 */
public class DwsTrafficItemPvUvWindow {
    public static void main(String[] args) throws Exception {
        // 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 读取商品详情页日志
        String topic = "dwd_traffic_item_detail_view";
        String groupId = "dws_traffic_item_pv_uv_window";
        SingleOutputStreamOperator<JSONObject> itemLogDS = env
                .addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId))
                .flatMap(new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                        out.collect(JSON.parseObject(value));
                    }
                });

        // 3. 提取事件时间
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = itemLogDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((json, ts) -> json.getLong("ts"))
        );

        // 4. 按商品ID+设备ID分组去重（复用DwdTrafficUniqueVisitorDetail的去重逻辑）
        KeyedStream<JSONObject, String> keyedDS = withWatermarkDS
                .keyBy(json -> {
                    String itemId = json.getJSONObject("page").getString("item");
                    String mid = json.getJSONObject("common").getString("mid");
                    return itemId + "_" + mid; // 商品+设备唯一标识
                });

        // 5. 计算UV（每日去重）和PV（直接计数）
        SingleOutputStreamOperator<TrafficItemPvUvBean> uvPvDS = keyedDS
                .flatMap(new RichFlatMapFunction<JSONObject, TrafficItemPvUvBean>() {
                    private ValueState<String> lastVisitState;

                    @Override
                    public void open(Configuration parameters) {
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();
                        ValueStateDescriptor<String> desc = new ValueStateDescriptor<>("last-visit", String.class);
                        desc.enableTimeToLive(ttlConfig);
                        lastVisitState = getRuntimeContext().getState(desc);
                    }

                    @Override
                    public void flatMap(JSONObject json, Collector<TrafficItemPvUvBean> out) throws Exception {
                        String itemId = json.getJSONObject("page").getString("item");
                        String curDt = DateFormatUtil.toDate(json.getLong("ts"));
                        String lastDt = lastVisitState.value();

                        // PV：每次访问都计数
                        long pv = 1L;
                        // UV：每日去重计数
                        long uv = (lastDt == null || !lastDt.equals(curDt)) ? 1L : 0L;

                        if (uv == 1) {
                            lastVisitState.update(curDt);
                        }

                        out.collect(new TrafficItemPvUvBean(
                                "", "", itemId, pv, uv, json.getLong("ts")
                        ));
                    }
                });

        // 6. 窗口聚合（复用DwsTrafficPageViewWindow的窗口逻辑）
        SingleOutputStreamOperator<TrafficItemPvUvBean> resultDS = uvPvDS
                .windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficItemPvUvBean>() {
                    @Override
                    public TrafficItemPvUvBean reduce(TrafficItemPvUvBean v1, TrafficItemPvUvBean v2) throws Exception {
                        v1.setPv(v1.getPv() + v2.getPv());
                        v1.setUv(v1.getUv() + v2.getUv());
                        return v1;
                    }
                }, new AllWindowFunction<TrafficItemPvUvBean, TrafficItemPvUvBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TrafficItemPvUvBean> values, Collector<TrafficItemPvUvBean> out) throws Exception {
                        TrafficItemPvUvBean bean = values.iterator().next();
                        bean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        bean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });

        // 7. 写入ClickHouse（复用DwsTradeProvinceOrderWindow的写入工具）
        resultDS.print(">>>>>>>>>>>");
        resultDS.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_traffic_item_pv_uv_window values(?,?,?,?,?,?)"
        ));

        env.execute("DwsTrafficItemPvUvWindow");
    }
}
